package cn.nie.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;


public class HbaseUtils {


    /**
     * HBASE 表名称
     */
    public  final String TABLE_NAME = "doc";
    /**
     * 列簇1 文章信息
     */
    public  final String COLUMNFAMILY_1 = "info1";
    /**
     * 列簇1中的列
     */
    public  final String COLUMNFAMILY_1_TITLE = "title";
    public  final String COLUMNFAMILY_1_AUTHOR = "author";
    public  final String COLUMNFAMILY_1_CONTENT = "content";
    public  final String COLUMNFAMILY_1_DESCRIBE = "describe";

    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;


    /**
     * 初始化方法（备用）
     * @return
     * @throws IOException
     */
    public Configuration init() throws IOException {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort","2181");
        configuration.set("hbase.zookeeper.quorum","master.hadoop,slave1.hadoop,slave2.hadoop");
        connection = ConnectionFactory.createConnection(configuration);
        return configuration;
    }
    /**
     * 构造函数加载配置
     */
    public HbaseUtils() {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort","2181");
        configuration.set("hbase.zookeeper.quorum","master.hadoop,slave1.hadoop,slave2.hadoop");
        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void createTable(String tableName,Configuration configuration) throws IOException {
        System.out.println("start create table ......");
        try {
            HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);
            if (hBaseAdmin.tableExists(tableName)) {// 如果存在要创建的表，那么先删除，再创建
                hBaseAdmin.disableTable(tableName);
                hBaseAdmin.deleteTable(tableName);
                System.out.println(tableName + " is exist,detele....");
            }
            HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
            tableDescriptor.addFamily(new HColumnDescriptor("column1"));
            tableDescriptor.addFamily(new HColumnDescriptor("column2"));
            tableDescriptor.addFamily(new HColumnDescriptor("column3"));
            hBaseAdmin.createTable(tableDescriptor);
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        System.out.println("end create table ......");
    }

    public  void  put(String tableName, String row, String columnFamily, String column, String data) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(row));
        put.addColumn(columnFamily.getBytes(),column.getBytes(),data.getBytes());
        table.put(put);
        System.out.println("put'" + row + "'," + columnFamily + ":" + column + "','" + data + "'");
    }

}
